package com.it.bigdata.zookeeper;

import com.it.bigdata.mail.MyMail;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.Test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class PathCacheTest {
    @Test
    public void testPathChildern() throws Exception {
        String path="/node11";
        //指定zk集群的地址
        String connectString = "node3:2181,node4:2181,node5:2181";
        // 重试策略： 1000 ：代表是重试时间间隔     3：表示是重试次数（连接zookeeper时使用）
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        //使用curator创建一个zk链接
        int sessionTimeoutMs = 2000;//这个值必须在4s--40s之间，表示是链接失效的时间
        int connectionTimeoutMs = 1000;//链接超时时间
        //使用curator框架提供的方法创建一个连接zk的对象
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(sessionTimeoutMs)
                .connectionTimeoutMs(connectionTimeoutMs)
                .retryPolicy(retryPolicy)
                .build();
        /**
         * 在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理
         *  ExecutorService pool = Executors.newFixedThreadPool(2);
         */

        //启动zk连接（开启事件监听）
        client.start();
        //构造一个缓存实例对象，参数分别为连接zk的对象，监听的节点路径，是否缓存节点内容
        //监听数据节点的增删改，会触发事件
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client,path,true);

        //PathChildrenCacheListener一次父节点注册，监听每次子节点操作，不监听自身和查询。
        //注册字节点监听器
        PathChildrenCacheListener pccl=new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        break;
                    case CHILD_UPDATED:
                        break;
                    case CHILD_REMOVED:
                        MyMail.sendMsg();
                        break;
                    default:
                        break;
                }
            }
        };
        //添加节点监听事件,缓存改变每次都会出发PathChildrenCacheListener
        pathChildrenCache.getListenable().addListener(pccl);
        /*这里的start方法，需要传入启动的模式。可以传入三种模式，也就是API列表中看到的StartMode，其中定义了下面三种枚举：
        （ 1）NORMAL——异步初始化cache
        （2）BUILD_INITIAL_CACHE——同步初始化cache
        （3）POST_INITIALIZED_EVENT——异步初始化cache，并触发完成事件*/
        //缓存使用同步初始化
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        System.in.read();
    }
}
